package service

import (
	"runtime/debug"
	"time"

	"git.cloud.inspur.com/ichain/ichain-sdk-go/pkg/common/logging"
	"git.cloud.inspur.com/ichain/ichain-sdk-go/pkg/common/options"
	"git.cloud.inspur.com/ichain/ichain-sdk-go/pkg/ichain/events/service/blockfilter"
	"git.cloud.inspur.com/ichain/ichain-sdk-go/pkg/ichain/events/service/dispatcher"
	"git.cloud.inspur.com/ichain/ichain-sdk-go/pkg/providers/ichain"
	"github.com/pkg/errors"
)

const (
	stopTimeout = 5 * time.Second
)

var logger = logging.NewLogger("ichainsdk/ichain")

type Dispatcher interface {
	Start() error
	EventCh() (chan<- interface{}, error)
	LatestBlockNum() uint64
}

// Service 实现了provider的eventservice接口
type Service struct {
	params
	dispatcher Dispatcher
}

func New(dispatcher Dispatcher, opts ...options.Opt) *Service {
	params := defaultParams()
	options.Apply(params, opts)

	return &Service{
		params:     *params,
		dispatcher: dispatcher,
	}
}

func (s *Service) Start() error {
	return s.dispatcher.Start()
}

func (s *Service) Stop() {
	eventCh, err := s.dispatcher.EventCh()
	if err != nil {
		logger.Warnf("Error stopping event service: %s", err)
		return
	}

	errCh := make(chan error, 1)
	eventCh <- dispatcher.NewStopEvent(errCh)

	select {
	case err = <-errCh:
		if err != nil {
			logger.Warnf("Error while stopping dispatcher: %s", err)
		}
	case <-time.After(stopTimeout):
		logger.Infof("Timed out waiting for dispatcher to stop")
	}
}

// Submit 向事件线程提交事件
// @param event
// @return error
func (s *Service) Submit(event interface{}) error {
	defer func() {
		if p := recover(); p != nil {
			logger.Warnf("panic while submitting event: %s", p)
			debug.PrintStack()
		}
	}()

	eventCh, err := s.dispatcher.EventCh()
	if err != nil {
		return errors.WithMessage(err, "Error submitting to event dispatcher")
	}
	eventCh <- event

	return nil
}

func (s *Service) Dispatcher() Dispatcher {
	return s.dispatcher
}

// RegisterBlockEvent 注册区块事件
// @param filter
// @return ichain.Registration
// @return <-chan
// @return error
func (s *Service) RegisterBlockEvent(filter ...ichain.BlockFilter) (ichain.Registration, <-chan *ichain.BlockEvent, error) {
	// 事件返回通道，有了对应的事件之后会通过该通道返回
	eventch := make(chan *ichain.BlockEvent, s.eventConsumerBufferSize)
	// 返回注册信息，由于注册是异步的，需要监听一个通道，同步等待
	regch := make(chan ichain.Registration)
	errch := make(chan error)

	blockFilter := blockfilter.AcceptAny
	if len(filter) > 1 {
		return nil, nil, errors.New("only one block filter may be specified")
	}

	if len(filter) == 1 {
		blockFilter = filter[0]
	}

	if err := s.Submit(dispatcher.NewRegisterBlockEvent(blockFilter, eventch, regch, errch)); err != nil {
		return nil, nil, errors.WithMessage(err, "error registering for block events")
	}

	select {
	case response := <-regch:
		return response, eventch, nil
	case err := <-errch:
		return nil, nil, err
	}
}

func (s *Service) RegisterFilteredBlockEvent() (ichain.Registration, <-chan *ichain.FilteredBlockEvent, error) {
	eventch := make(chan *ichain.FilteredBlockEvent, s.eventConsumerBufferSize)
	regch := make(chan ichain.Registration)
	errch := make(chan error)

	if err := s.Submit(dispatcher.NewRegisterFilteredBlockEvent(eventch, regch, errch)); err != nil {
		return nil, nil, errors.WithMessage(err, "error registering for filtered block events")
	}

	select {
	case response := <-regch:
		return response, eventch, nil
	case err := <-errch:
		return nil, nil, err
	}
}

// RegisterContractEvent 注册合约事件
// @param ccID
// @param eventFilter
// @return ichain.Registration
// @return <-chan
// @return error
func (s *Service) RegisterContractEvent(ccID, eventFilter string) (ichain.Registration, <-chan *ichain.ContractEvent, error) {
	if ccID == "" {
		return nil, nil, errors.New("contract ID is required")
	}
	if eventFilter == "" {
		return nil, nil, errors.New("event filter is required")
	}

	eventch := make(chan *ichain.ContractEvent, s.eventConsumerBufferSize)
	regch := make(chan ichain.Registration)
	errch := make(chan error)

	if err := s.Submit(dispatcher.NewRegisterContractEvent(ccID, eventFilter, eventch, regch, errch)); err != nil {
		return nil, nil, errors.WithMessage(err, "error registering for contract events")
	}

	select {
	case response := <-regch:
		return response, eventch, nil
	case err := <-errch:
		return nil, nil, err
	}
}

// RegisterTxStatusEvent 注册交易状态事件
// @param txID
// @return ichain.Registration
// @return <-chan
// @return error
func (s *Service) RegisterTxStatusEvent(txID string) (ichain.Registration, <-chan *ichain.TxStatusEvent, error) {
	if txID == "" {
		return nil, nil, errors.New("txID must be provided")
	}

	eventch := make(chan *ichain.TxStatusEvent, s.eventConsumerBufferSize)
	regch := make(chan ichain.Registration)
	errch := make(chan error)

	if err := s.Submit(dispatcher.NewRegisterTxStatusEvent(txID, eventch, regch, errch)); err != nil {
		return nil, nil, errors.WithMessage(err, "error registering for Tx Status events")
	}

	select {
	case response := <-regch:
		return response, eventch, nil
	case err := <-errch:
		return nil, nil, err
	}
}

func (s *Service) Unregister(reg ichain.Registration) {
	if err := s.Submit(dispatcher.NewUnregisterEvent(reg)); err != nil {
		logger.Warnf("Error unregistering: %s", err)
	}
}
